#define FUSE_USE_VERSION 26

#ifdef HAVE_CONFIG_H
#include <config.h>
#endif

#ifdef linux
/* For pread()/pwrite() */
#define _XOPEN_SOURCE 500
#endif

#include "client.h"

uint32_t localip;
OSD *osd_head = NULL;
META_CACHE *meta_list = NULL;
pthread_mutex_t metalistlock = PTHREAD_MUTEX_INITIALIZER;
char host[INET_ADDRSTRLEN];
int TS = 1024 * 1024 * 8;//暂时指定，以后应该在初始化时通过MDS获取
static int dnfs_getattr(const char *path, struct stat *stbuf)
{
PDEBUG("[dnfs_getattr]: %s\n", path);
	CLIENT	*cls;
	lstat_out res;
	lstat_in arg;
	int ret = 0;
	arg.filename = path;

	pthread_mutex_lock(&metalistlock);//lock the whole meta list

    	cls = clnt_create (host, CS, CS_VERS, "udp");
	if (cls == NULL) {
		clnt_pcreateerror (host);
		//exit (1);
		ret = -1;
		goto out;
	}

	if (cs_lstat_2(&arg, &res, cls) != RPC_SUCCESS) {

		printf("%s\n", clnt_sperror(cls, host));
		ret = -2;
		goto out;
	}

	clnt_destroy(cls);
	ret = res.res;
	//把rpc数据类型转换成本地类型，这样可以在不同平台下操作，比如32位和64位之间
	stbuf->st_dev = res.stbuf.st_dev;
	stbuf->st_ino  = res.stbuf.st_ino;
	stbuf->st_mode = res.stbuf.st_mode;
	stbuf->st_nlink= res.stbuf.st_nlink;
	stbuf->st_uid = res.stbuf.st_uid;
	stbuf->st_gid = res.stbuf.st_gid;
	stbuf->st_rdev = res.stbuf.st_rdev;
	stbuf->st_size = res.stbuf.st_size;
	stbuf->st_blksize = res.stbuf.st_blksize;
	stbuf->st_blocks = res.stbuf.st_blocks;
	stbuf->st_atime = res.stbuf.atime;
	stbuf->st_mtime = res.stbuf.mtime;
	stbuf->st_ctime = res.stbuf.ctime;
//	print_stat(stbuf, path);
	pthread_mutex_unlock(&metalistlock);//lock the whole meta list
out:
	return ret;
}

static int dnfs_access(const char *path, int mask)
{
    	CLIENT	*cls;
	int res;
	PDEBUG("[dnfs_access]: %s\n", path);
	access_in arg;

	arg.filename = path;
	arg.mask = mask;
	cls = clnt_create (host, CS, CS_VERS, "udp");
	if (cls == NULL) {
		clnt_pcreateerror (host);
		return 1;
	}

    	if (cs_access_2(&arg, &res, cls) != RPC_SUCCESS) {
		printf("%s\n", clnt_sperror(cls, host));
		return -1;
	}
	clnt_destroy(cls);
	return res;
}


static int dnfs_readdir(const char *path, void *buf, fuse_fill_dir_t filler, off_t offset, struct fuse_file_info *fi)
{
	PDEBUG("[dnfs_readdir]: %s\n", path);	
    	CLIENT	*cls;
	DIR *dp;
	opendir_in  openin;
	opendir_out openres;
	readdir_in  readin;
	readdir_out readres;
	(void) offset;
    	(void) fi;
	
	openin.filename = path;
	
	cls = clnt_create (host, CS, CS_VERS, "udp");
	if (cls == NULL) {
		clnt_pcreateerror (host);
		return -1;//直接返回，不需任何回收处理
	}

	if (cs_opendir_2(&openin, &openres, cls) != RPC_SUCCESS) {
		printf("%s\n", clnt_sperror(cls, host));
		return -1;
	}
	dp = (long)openres.dp;
//PDEBUG("dp = %p\n", dp);
	if (dp == NULL) {
		PDEBUG("dp = NULL, return\n");
		return openres.res;
	}
	readin.dp = (long)dp;
	readres.d_name = (char *) malloc (NAMELEN);
	while (1) {
		if (cs_readdir_2(&readin, &readres, cls) != RPC_SUCCESS) {
			printf("%s\n", clnt_sperror(cls, host));
			free(readres.d_name);
			return -1;
		}

		if (readres.d_ino == 0 && readres.d_type == 0) {
			PDEBUG("readdir have reach the end\n");
			break;
		}	
		
		struct stat st;
        	memset(&st, 0, sizeof(st));	      
        	st.st_ino = readres.d_ino;
        	st.st_mode = readres.d_type << 12;
        	if (filler(buf, readres.d_name, &st, 0)) {
			PDEBUG("filler break, d_name = %s\n", readres.d_name);
			break;
		}	
	}
	free(readres.d_name);
	if (cs_closedir_2(&readin, NULL, cls) != RPC_SUCCESS) {
		printf("%s\n", clnt_sperror(cls, host));
		return -1;	
	}
	clnt_destroy(cls);
    	return 0;
}


static int dnfs_mkdir(const char *path, mode_t mode)
{
    	PDEBUG("[dnfs_mkdir]: %s\n", path);
    	int res;
	CLIENT	*cls;
	mkdir_in in;

	in.filename = path;
	in.mode = mode;
//PDEBUG("mode = %d\n", mode);

	cls = clnt_create (host, CS, CS_VERS, "udp");
	if (cls == NULL) {
		clnt_pcreateerror (host);
		return -1;//直接返回，不需任何回收处理
	}
	if (cs_mkdir_2(&in, &res, cls) != RPC_SUCCESS) {
		printf("%s\n", clnt_sperror(cls, host));
		return -1;
	}
	clnt_destroy(cls);

	return res;
}


static int dnfs_rmdir(const char *path)
{
	PDEBUG("[dnfs_rmdir]: %s\n", path);
    	int res;
	CLIENT	*cls;
	rmdir_in in;

	in.filename = path;

	cls = clnt_create (host, CS, CS_VERS, "udp");
	if (cls == NULL) {
		clnt_pcreateerror (host);
		return -1;//直接返回，不需任何回收处理
	}
	if (cs_rmdir_2(&in, &res, cls) != RPC_SUCCESS) {
		printf("%s\n", clnt_sperror(cls, host));
		return -1;
	}
	clnt_destroy(cls);

    	return res;
}

static int dnfs_mknod(const char *path, mode_t mode, dev_t rdev)
{
	PDEBUG("[dnfs_mknod] : %s\n", path);
	CLIENT	*cls;
	int res;
	mknod_in arg;
   
	arg.filename = path;
	arg.rdev = rdev;
	arg.mode = mode;

//PDEBUG("mode = %o, rdev = %ld\n", mode, (long)rdev);   	
	cls = clnt_create (host, CS, CS_VERS, "udp");
	if (cls == NULL) {
		clnt_pcreateerror (host);
		return -1;//直接返回，不需任何回收处理
	}
	
	if (cs_mknod_2(&arg, &res, cls) != RPC_SUCCESS) {
		printf("%s\n", clnt_sperror(cls, host));
		return -1;
	}
PDEBUG("res = %d\n", res);

	clnt_destroy(cls);
	return res;
}


static int dnfs_open(const char *path, struct fuse_file_info *fi)
{
	PDEBUG("[dnfs_open]: %s, mode = %#X\n", path, fi->flags);
	CLIENT	*cls;
	open_in arg;
	open_out res;
	META_CACHE *meta = NULL;
	arg.filename = path;
	int ret = 0;

	pthread_mutex_lock(&metalistlock);//lock the whole meta list
	
	if ((meta = find_meta(path)) != NULL) {//已经打开过，这时意味着有多个进程打开该文件
	//	pthread_mutex_lock(&meta->lock);
		if ((fi->flags & 0xF) == O_RDONLY) {//以只读打开,并且当前没有进程写该文件
			meta->rcount++;
		}
		else if((fi->flags & 0xF) == O_WRONLY || (fi->flags & 0xF) == O_RDWR) {
			if (meta->wcount > 0) {//表明本client已获得该文件的写权限
				meta->wcount++;
			}
			else {//本client没获得该文件的写权限， 需要把读权限升级为写权限
				PDEBUG("need to upgrade write permission now\n");
				ret = update_permission(path, RTOW);
				if (ret == 0)//succefully 
					meta->wcount++;				
			}					
		}
		else {
			printf("flags illeagle:%#X\n", fi->flags);
			ret = -1;
		}
	//	pthread_mutex_unlock(&meta->lock);
		goto out;
	}

	//该文件目前没被打开
	arg.mode = fi->flags;
	arg.ip = localip;
	//meta_data could be very large (5G's file will contain more than 17K's meta info), so use tcp here
	cls = clnt_create (host, CS, CS_VERS, "tcp");
	if (cls == NULL) {
		clnt_pcreateerror (host);
		ret = -1;		
		goto out;//直接返回，不需任何回收处理
	}
	
	res.buf.buf_val = (char *) malloc (META_SIZE);
	if (res.buf.buf_val == NULL) {
		fprintf(stderr, "malloc res.buf.buf_val return NULL\n");
		exit(-2);
		goto out;
	}

	if (cs_open_2(&arg, &res, cls) != RPC_SUCCESS) {
		printf("%s\n", clnt_sperror(cls, host));
		clnt_destroy(cls);
		ret = -1;
		goto dofree;
	}
	clnt_destroy(cls);
	ret = res.res;
	if (ret != 0) {
		printf("cs_open failed, ret = %d\n", ret);
		goto out;
	}
	meta = (META_CACHE *) malloc (sizeof(META_CACHE));
	meta->path = (char *) malloc (strlen(path) + 1);
	snprintf(meta->path, strlen(path) + 1, "%s", path);
	if ((fi->flags & 0xF) == O_RDONLY) {//以只读打开
		meta->rcount = 1;
		meta->wcount = 0;
	}
	else if((fi->flags & 0xF) == O_WRONLY || (fi->flags & 0xF) == O_RDWR) {
		meta->wcount = 1;
		meta->rcount = 0;
	}

	int err;
	if ((err = pthread_mutex_init(&meta->lock, NULL)) != 0) {
		printf("**********init lock failed :%s************\n", strerror(err));
		if (meta->path != NULL) {
			free(meta->path);
		}
		if (meta != NULL) {
			free(meta);
		}
		goto dofree;
	}
	meta->newsize = meta->oldsize = res.filesize;
	meta->isdirty = 0;//not dirty now
	getchunk(res.buf.buf_val, res.buf.buf_len, meta);
	//print_one_metacache(meta);
	insert_meta(meta);
dofree:
	free(res.buf.buf_val);
out:
//	print_all_metacache();	
	pthread_mutex_unlock(&metalistlock);//unlock the whole meta list

PDEBUG("[open]res = %d\n", ret);
	return ret;  //正常时res为0；否则res < 0
}


static int dnfs_release(const char *path, struct fuse_file_info *fi)
{
	PDEBUG("[dnfs_release]: \"%s\", flags = %#X\n", path, fi->flags);
    	(void) path;
    	(void) fi;
	CLIENT	*cls;
	int res = -1;
	int flags = fi->flags;
	cs_release_in cs_in;
	cs_in.path = path;
	cs_in.buf.buf_val = NULL;

	pthread_mutex_lock(&metalistlock);//lock list
	META_CACHE *meta = find_meta(path);
	if (meta == NULL) {
		printf("Can't find meta info of %s\n", path);
		exit(1);
	}

	if ((flags & 0xF) == O_RDONLY) {//以只读打开
		meta->rcount--;//把读引用数减1
	}
	else if ((flags & 0xF) == O_WRONLY || (flags & 0xF) == O_RDWR) {//以写打开
		meta->wcount--;//把写引用数减1
		if (meta->wcount == 0 && meta->rcount > 0) {
			//down write permission to read permission here
			if ((res = update_permission(path, WTOR)) != 0) {
				pthread_mutex_unlock(&metalistlock);//unlock list
				return res;
			}
		}
	}
	
	if (meta->rcount > 0 || meta->wcount > 0) {
		//pthread_mutex_unlock(&);
		PDEBUG("rcount = %d, wcount = %d\n", meta->rcount, meta->wcount);
		pthread_mutex_unlock(&metalistlock);//unlock list
		return 0;
	}
	cs_in.ip = localip;
	cs_in.isdirty = meta->isdirty;
	if (meta->isdirty == 0) {//meta->rcount == 0 && meta->wcount == 0 
		cs_in.newsize = 0;
		cs_in.buf.buf_val = (char *) malloc (1);
		cs_in.buf.buf_len = 0;
	}
	else if (meta->isdirty == 1) {//meta->rcount == 0 && meta->wcount == 0
		cs_in.newsize = meta->newsize;
		cs_in.buf.buf_len = meta->n_chunknum * ICHUNK_SIZE;
		PDEBUG("buf_len = %u\n", cs_in.buf.buf_len);
		cs_in.buf.buf_val = (char *) calloc (1, cs_in.buf.buf_len);
		char *index = cs_in.buf.buf_val;
		int i;
		for (i = 0; i < meta->n_chunknum; i++) {
			if (meta->chunkarray[i] != NULL)//如果==NULL，则不管，保持为0
				memcpy(index, meta->chunkarray[i], ICHUNK_SIZE);
			index += ICHUNK_SIZE;
		}
		CHUNK_MEM* tmp = (CHUNK_MEM*)cs_in.buf.buf_val;
		for(i = 0; i < meta->n_chunknum; i++) {
			PDEBUG("%s\t%u\t%u\n", tmp->oid, tmp->ip, tmp->size);
			tmp ++;
		}
	}

	cls = clnt_create (host, CS, CS_VERS, "tcp");
	if (cls == NULL) {
		clnt_pcreateerror (host);
		goto out;
	}
	if (cs_release_2(&cs_in, &res, cls) != RPC_SUCCESS) {
		printf("%s\n", clnt_sperror(cls, host));
		clnt_destroy(cls);
		goto out;
	}
	clnt_destroy(cls);
	
	if (res != 0) {
		printf("cs_release failed\n");
	}
out:
	if (cs_in.buf.buf_val != NULL) 
		free(cs_in.buf.buf_val);
PDEBUG("Before delete: %s\n", path);
	print_all_metacache();
	//pthread_mutex_unlock(&);
	delete_meta(path);
PDEBUG("After delete: %s\n", path);
	print_all_metacache();

	pthread_mutex_unlock(&metalistlock);//uplock	
PDEBUG("[release]res = %d\n", res);
	return res;
}

/*
 * 
 */
static int dnfs_write(const char *path, const char *buf, size_t size,
                     off_t offset, struct fuse_file_info *fi)
{
	PDEBUG("[dnfs_write]: %s, size = %ld, offset = %ld\n", path, size, offset);
	(void) fi;
	int ret = -1;

	META_CACHE *meta;

	pthread_mutex_lock(&metalistlock);//lock the whole meta list
	if ((meta = find_meta(path)) == NULL) {//出现这种情况有可能是因为一个进程在写，另外一个进程把该文件release掉了或者write之前没有open（不过这种情况有可能出现吗？）
		pthread_mutex_unlock(&metalistlock);//uplock the whole meta list
		printf("Can't find the meta_cache of %s\n", path);
		exit(-1);
		//return -1;
	}
	pthread_mutex_unlock(&metalistlock);//uplock the whole meta list

//	pthread_mutex_lock(&meta->lock);
//PDEBUG("********writlock********\n");
	//o1,o2是需要write的第一个分条号和最后一个分条号
	int o1, o2;
	o1 = getcid(offset);
//	o1 = ceil(offset / TS);
	o2 = getcid(offset + size - 1);
//	PDEBUG("o1 = %d, o2 = %d\n", o1, o2);
	int i;
	int count, res;
	res = 0;
	co_write_in	co_in;
	uint32_t written_size = 0;
	OSD *osd = NULL;
	for (i = o1; i <= o2; i++) {
		if (i >= meta->n_chunknum) {
			PDEBUG("print_one_metacache before add_chunk\n");
			print_one_metacache(meta);
			if (ask_newchunk(path, i, meta) == -1) {
				fprintf(stderr, "ask_newchunk failed 1\n");
				goto out;
			}
			PDEBUG("print_one_metacache after add_chunk\n");
			print_one_metacache(meta);
		}
		else if (meta->chunkarray[i] == NULL) {//该chunk不存在，比如空洞,这时该分条的元数据还没分配
			PDEBUG("print_one_metacache before add_chunk\n");
			print_one_metacache(meta);
			if (ask_newchunk(path, i, meta) == -1) {
				fprintf(stderr, "ask_newchunk failed 2\n");
				goto out;
			}
			PDEBUG("print_one_metacache after add_chunk\n");
			print_one_metacache(meta);
		}
		if ((osd = get_osd(meta->chunkarray[i]->ip)) == NULL) {
			ret = -4;
			goto out;
		}
		co_in.oid = meta->chunkarray[i]->oid;
		co_in.buf.buf_val = buf + written_size;
		if (o1 == o2) {//要写的数据只在同一oss中
			co_in.offset = offset % TS;
			co_in.buf.buf_len = size;		
//			PDEBUG("write status 1: 1 oss, offset = %lu, size = %u\n", co_in.offset, co_in.buf.buf_len);
		}
		else if (i == o1) {// && i != o2) {//多块中的第一块
			PDEBUG("write status 2\n");
			co_in.offset = offset % TS;
			co_in.buf.buf_len = TS - co_in.offset;//即剩下的空间
		}
		else if (i == o2) {//多块中的最后一块
			PDEBUG("write status 3\n");
			co_in.offset = 0;
			co_in.buf.buf_len = (size + offset) % TS;
			if (co_in.buf.buf_len == 0)//这时刚好整块，而不是0
				co_in.buf.buf_len = TS;
		}
		else if (i > o1 && i < o2) {//多块中的中间那些块.由于TS远远大于每次I/O请求的大小，所以不可能到这这种情况
			PDEBUG("write status 4\n");
			co_in.offset = 0;
			co_in.buf.buf_len = TS;
		}
		else {//shoule never reach here
			PDEBUG("write errno\n");
			ret = -2;
			goto out;
		}

		pthread_mutex_lock(&osd->lock);
		int err = co_write_2(&co_in, &count, osd->cl);
		if (err != RPC_SUCCESS) {
			char ipp[INET_ADDRSTRLEN];
			clnt_perror(osd->cl, inet_ntop(AF_INET, &meta->chunkarray[i]->ip, ipp, INET_ADDRSTRLEN));
			if (err == RPC_CANTSEND || err == ECONNRESET){// == 3,这时与osd的连接已经失效，重连
				PDEBUG("re-create osd handle\n");
				sleep(2);
				CLIENT *tmpcl = create_osd_handle(osd->ip);//重新创建连接 
				if (tmpcl == NULL) {
					pthread_mutex_unlock(&osd->lock);
					goto out;
				}
				clnt_destroy(osd->cl);//销毁原来的连接
				osd->cl = tmpcl;

				if (co_write_2(&co_in, &count, osd->cl) != RPC_SUCCESS) {
					pthread_mutex_unlock(&osd->lock);
					printf("faild again\n");
					clnt_perror(osd->cl, inet_ntop(AF_INET, &meta->chunkarray[i]->ip, ipp, INET_ADDRSTRLEN));
					goto out;
				}
			}
			else { //does not handle other error here
				pthread_mutex_unlock(&osd->lock);
				goto out;
			}
		}
		pthread_mutex_unlock(&osd->lock);
		//co_write调用成功
		if (count == co_in.buf.buf_len) {//write成功
			written_size += count;
		}
		else {
			printf("count = %d, but expect %d\n", count, co_in.buf.buf_len);
			ret = -1;
			goto out;
		}
		if (co_in.offset + co_in.buf.buf_len > meta->chunkarray[i]->size) {//修改chunk的大小，用于文件空洞
			meta->chunkarray[i]->size = co_in.offset + co_in.buf.buf_len;
		}
	}
	ret = written_size;
	if (written_size + offset > meta->newsize) {//修改文件的总大小
		meta->isdirty = 1;
		meta->newsize = written_size + offset;
	}
out:
//	pthread_mutex_unlock(&meta->lock);//unlock meta->list
//PDEBUG("********writunlock********\n");	
PDEBUG("[write]ret = %d\n", ret);
	return ret;
}


static int dnfs_read(const char *path, char *buf, size_t size, off_t offset,
                    struct fuse_file_info *fi)
{
	PDEBUG("[dnfs_read]: %s, size = %ld, offset = %ld\n", path, size, offset);
	int ret = 0;
	uint32_t read_size = 0;
	(void) fi;
	co_read_in	co_in;
	co_read_out	co_out;

	META_CACHE *meta;
	pthread_mutex_lock(&metalistlock);//lock the whole meta list
	if ((meta = find_meta(path)) == NULL) {
		pthread_mutex_unlock(&metalistlock);//uplock the whole meta list
		printf("Can't find the meta_cache of %s\n", path);
		exit -1;
		//return -1;
	}
	pthread_mutex_unlock(&metalistlock);//uplock the whole meta list

//	pthread_mutex_lock(&meta->lock);
//PDEBUG("********readlock********\n");
	if (offset >= meta->newsize) {
		PDEBUG("read reach the end\n");
		//return 0;//0表示读到终点，不能再读
		ret = 0;
		goto out;
	}
	if (offset + size > meta->newsize) {//size过大，已超过文件结尾，修改size
		size = meta->newsize - offset;
		PDEBUG("Read over the file, set size = %lu\n", size);
	}

	//o1,o2是需要read的第一个分条号和最后一个分条号
	int o1, o2;
	o1 = getcid(offset);
	//o1 = ceil(offset / TS);
	o2 = getcid(offset + size - 1);
//	PDEBUG("o1 = %d, o2 = %d\n", o1, o2);
	int i;
	int count = 0;
	OSD *osd = NULL;
	for (i = o1; i <= o2; i++, read_size += co_in.size) {
		if (o1 == o2) {//要写的数据只在同一oss中
			co_in.offset = offset % TS;
			co_in.size = size;		
	//		PDEBUG("read status 1: 1 oss, offset = %lu, size = %u\n", co_in.offset, co_in.size);
		}
		else if (i == o1) {// && i != o2) {//多块中的第一块
			PDEBUG("read status 2\n");
			co_in.offset = offset % TS;
			co_in.size = TS - co_in.offset;//即剩下的空间
		}
		else if (i == o2) {//多块中的最后一块
			PDEBUG("read status 3\n");
			co_in.offset = 0;
			co_in.size = (size + offset) % TS;
			if (co_in.size == 0)//这时刚好整块，而不是0
				co_in.size = TS;
		}
		else if (i > o1 && i < o2) {//多块中的中间那些块.由于TS远远大于每次I/O请求的大小，所以不可能到这这种情况
			PDEBUG("read status 4\n");
			co_in.offset = 0;
			co_in.size = TS;
		}
		else {//shoule never reach here
			PDEBUG("read errno\n");
			ret = -2;
			goto out;

		}
		count = co_in.size;//count标记了本次填充到buf中的数据字节数（包括本地填充的空洞）
		if (meta->chunkarray[i] == NULL || meta->chunkarray[i]->size <= co_in.offset) {//该chunk不存在，或者所需信息全部处于chunk最后的空洞中，这时只需把本地buf相应地方填0
			PDEBUG("chunk_mem is not exist, File Hole here: pid = %d\n", i);
			bzero(buf + read_size, co_in.size);
			continue;
		}
		if (meta->chunkarray[i]->size < co_in.offset + co_in.size) {//所需数据部分处在空洞中
			PDEBUG("Part of the data is in the File Hole: pid = %d\n", i);
			bzero(buf + read_size + meta->chunkarray[i]->size - co_in.offset,
			co_in.size + co_in.offset - meta->chunkarray[i]->size);
			//改变size
			co_in.size = meta->chunkarray[i]->size - co_in.offset;
		}
		//前面已经把offset和size域准备好了，这里还有oid要得到
		co_in.oid = meta->chunkarray[i]->oid;
		co_out.buf.buf_val = buf + read_size;
		if ((osd = get_osd(meta->chunkarray[i]->ip)) == NULL) {
			ret = -4;
			goto out;
		}

		pthread_mutex_lock(&osd->lock);
		int err = co_read_2(&co_in, &co_out, osd->cl);
		if (err != RPC_SUCCESS){
			char ipp[INET_ADDRSTRLEN];
			clnt_perror(osd->cl, inet_ntop(AF_INET, &osd->ip, ipp, INET_ADDRSTRLEN));
			if (err == RPC_CANTSEND || err == ECONNRESET){// == 3,这时与osd的连接已经失效，重连
				sleep(5);
				PDEBUG("re-create osd handle\n");
				CLIENT *tmpcl = create_osd_handle(osd->ip);//重新创建连接 
				if (tmpcl == NULL) {
					pthread_mutex_unlock(&osd->lock);
					ret = -5;
					goto out;
				}
				clnt_destroy(osd->cl);//销毁原来的连接
				osd->cl = tmpcl;
				if (co_read_2(&co_in, &co_out, osd->cl) != RPC_SUCCESS) {
					pthread_mutex_unlock(&osd->lock);
					PDEBUG("faild again\n");
				//	clnt_perror(osd->cl, osd->ip);
					ret = -5;
					goto out;
				}
			}
			else { //does not handle other error here
				pthread_mutex_unlock(&osd->lock);
				ret = -5;
				goto out;
			}
		}
		pthread_mutex_unlock(&osd->lock);
		if (co_out.res != co_in.size) {//正确时res为写入的Byte数，应等于co_in.size;出错时为负数
			printf("[co_read] %u : co_out.res = %d, co_in.size = %u\n", osd->ip, co_out.res, co_in.size);
			ret = co_out.res;		
			exit (-6);//for debug
			goto out;
		}
	}
out:
//	pthread_mutex_unlock(&meta->lock);
//PDEBUG("********readunlock********\n");	
	if (ret == 0)//正确的情况下
		ret = read_size;
PDEBUG("[read]ret = %d\n", ret);
	return ret;
}


/*
 * 锁的粒度是否过大？
 */
static int dnfs_truncate(const char *path, off_t size)
{
    	PDEBUG("[dnfs_truncate]: %s, size %ld\n", path, size);
    	int res;
	int ret = 0;
	CLIENT	*cls;
	cs_truncate_in cs_in;
	cs_in.buf.buf_val = NULL;
	int isopend = 1;

	META_CACHE *meta;
	pthread_mutex_lock(&metalistlock);
	meta = find_meta(path);
	if (meta == NULL) {
		isopend = 0;//not opend now
		PDEBUG("%s is not opend now.\n", path);
	}
//	pthread_mutex_unlock(&metalistlock);

	/*准备cs_in阶段*/
/*	if (isopend) { 
		pthread_mutex_lock(&meta->lock);//lock here!!!!!!
PDEBUG("********truncatelock********\n");
	}		
*/	if (isopend && meta->isdirty) {
		cs_in.isdirty = 1;
		cs_in.newsize = meta->newsize;
		cs_in.buf.buf_len = meta->n_chunknum * ICHUNK_SIZE;
		PDEBUG("buf_len = %u\n", cs_in.buf.buf_len);
		cs_in.buf.buf_val = (char *) malloc (cs_in.buf.buf_len);
		char *index = cs_in.buf.buf_val;
		int i;
		for (i = 0; i < meta->n_chunknum; i++) {
			memcpy(index, meta->chunkarray[i], ICHUNK_SIZE);
			index += ICHUNK_SIZE;
		}
		CHUNK_MEM* tmp = (CHUNK_MEM*)cs_in.buf.buf_val;
		for(i = 0; i < meta->n_chunknum; i++) {
			PDEBUG("%s\t%u\t%u\n", tmp->oid, tmp->ip, tmp->size);
			tmp ++;
		}
	}
	else {//not opend or is not dirty
		cs_in.isdirty = 0;
		cs_in.newsize = 0;//nonsense here
		cs_in.buf.buf_val = (char *) malloc (1);
		cs_in.buf.buf_len = 0;
	}
	cs_in.filename = path;
	cs_in.truncate_size = size;
	cs_in.ip = localip;
	//传输阶段，无论是否opened，都是一样
	cls = clnt_create (host, CS, CS_VERS, "tcp");
	if (cls == NULL) {
		clnt_pcreateerror (host);
		ret = -1;
		goto out;
	}
	
	if (cs_truncate_2(&cs_in, &res, cls) != RPC_SUCCESS) {
		fprintf(stderr, "%s\n", clnt_sperror(cls, host));
		ret = res;
		goto out;
	}	
	clnt_destroy(cls);

	if (res < 0) {
		ret = res;
		goto out;
	}
	//改变本地元数据阶段——只有opened才需要）
	if (isopend) {
printf("before truncate:\n");
print_one_metacache(meta);
		//计算得到新的chunknum
		CHUNKNUM nchunknum;
		if (size == 0)
			nchunknum = 0;
		else 
			nchunknum = getcid(size - 1) + 1;
printf("nchunknum = %d, meta->n_chunknum = %d\n", nchunknum, meta->n_chunknum);
		if (nchunknum < meta->n_chunknum) {
			int i;
			for (i = nchunknum; i < meta->n_chunknum; i++) {
				if (meta->chunkarray[i] != NULL) {
					free(meta->chunkarray[i]);
					meta->chunkarray[i] = NULL;
				}
			}
			meta->n_chunknum = nchunknum;
		}
		if (size < meta->newsize && size != 0)//变小且不为0，变大不需要
			meta->chunkarray[nchunknum - 1]->size = size % TS;
		meta->newsize = size;
		meta->isdirty = 0;//上面已经刷新过
printf("after truncate:\n");
print_one_metacache(meta);
	}
out:
/*	if (isopend)  {
		pthread_mutex_unlock(&meta->lock);
PDEBUG("********truncateunlock********\n");
	}
*/	if (cs_in.buf.buf_val != NULL) 
		free(cs_in.buf.buf_val);
PDEBUG("[truncate]res = %d\n", ret);
	pthread_mutex_unlock(&metalistlock);

	return ret;
}


static int dnfs_unlink(const char *path)
{
	PDEBUG("[dnfs_unlink]: %s.", path);
    	int res;
	CLIENT	*cls;
	cs_unlink_in arg;

	arg.filename = path;
	arg.ip = localip;
	cls = clnt_create (host, CS, CS_VERS, "udp");
	if (cls == NULL) {
		clnt_pcreateerror (host);
		return -1;//直接返回，不需任何回收处理
	}
	
	if (cs_unlink_2(&arg, &res, cls) != RPC_SUCCESS) {
		printf("%s\n", clnt_sperror(cls, host));
		return -1;
	}
	
	clnt_destroy(cls);
PDEBUG("res = %d\n", res);

	return res;
}


static int dnfs_rename(const char *from, const char *to)
{
    	int res;
	int ret;
	PDEBUG("[rename]from %s to %s\n", from, to);
	META_CACHE *meta;
	const char *path = from;

	pthread_mutex_lock(&metalistlock);
	meta = find_meta(path);
	if (meta != NULL) {
		PDEBUG("%s is opend now.\n", path);
		ret = -EAGAIN;
		goto out;
	}
	else {
		CLIENT	*cls;
		cs_rename_in cs_in;
		cs_in.from = from;
		cs_in.to = to;
		cls = clnt_create (host, CS, CS_VERS, "udp");
		if (cls == NULL) {
			clnt_pcreateerror (host);
			ret = -1;
			goto out;
		}
	
		if (cs_rename_2(&cs_in, &res, cls) != RPC_SUCCESS) {
			printf("%s\n", clnt_sperror(cls, host));
			ret = -1;
			goto out;
		}
	
		clnt_destroy(cls);
		ret = res;
	}

out:
	pthread_mutex_unlock(&metalistlock);
PDEBUG("ret = %d\n", ret);
   	return ret;
}


static struct fuse_operations dnfs_oper = {
    	.getattr	= dnfs_getattr,
	.access		= dnfs_access,
	.readdir	= dnfs_readdir,
	.mknod		= dnfs_mknod,
	.open		= dnfs_open,
	.write		= dnfs_write,
	.unlink		= dnfs_unlink,
	.truncate	= dnfs_truncate,
	.release	= dnfs_release,
	.read		= dnfs_read,
	.mkdir		= dnfs_mkdir,
	.rmdir		= dnfs_rmdir,
	.rename		= dnfs_rename,
};

int main(int argc, char *argv[])
{
	PDEBUG("Use local CLIENT\n");
	umask(0);
	if (init() != 0) {
		return -1;
	}
/*
	struct rlimit r;
    	int resource=RLIMIT_NOFILE;
    	printf("the original OPEN_MAX=%ld,",sysconf(_SC_OPEN_MAX));
    	r.rlim_cur=OPEN_MAX;
    	r.rlim_max=OPEN_MAX;
    	if(setrlimit(resource,&r)!=0) {
    		printf("Set OPEN_MAX failed.Make sure you are ROOT!\n");
        	return -1;
    	}
	printf("set it to=%ld %ld\n",r.rlim_cur,r.rlim_max);
*/
//	printf("host: %s\n", host);
//	PDEBUG("host: %s\n", host);
/*	cls = clnt_create (host, CS, CS_VERS, "tcp");
	if (cls == NULL) {
		clnt_pcreateerror (host);
		//exit (1);
		return 1;
	}
*/
/*	cls = create_mds_handle(host);
	if (cls == NULL) {
		return -1;
	}
*/	
    	return fuse_main(argc, argv, &dnfs_oper, NULL);
}


 
 
